package cn.cup.dmp.etl


import cn.cup.dmp.beans.LogSchema
import cn.cup.dmp.config.ConfigHelper
import cn.cup.dmp.utils.NumParse
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Row, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}


object Bzio2Partique {
  /**
    * 将bz2文件转化为parquet文件
    */
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("DMP平台").setMaster("local[*]")
    conf.set("spark.serializer",ConfigHelper.ser)//设置序列化方式
    val context = new SparkContext(conf)
    val sql = new SQLContext(context)
      sql.setConf("spark.sql.parquet.compression.codec", "snappy")
    //读取文件
    val file = context.textFile(ConfigHelper.srcfilePath)
    //转化为parquet dataFrame=> parquet
    val legalLogs: RDD[Array[String]] = file.map(line => {
      line.split(",",-1)//防止后面为空，不进行切分的问题
    }).filter(_.length>=85)

    val value: RDD[Row] = legalLogs.map(arr =>Row(
        arr(0),
        NumParse.toInt(arr(1)),
        NumParse.toInt(arr(2)),
        NumParse.toInt(arr(3)),
        NumParse.toInt(arr(4)),
        arr(5),
        arr(6),
        NumParse.toInt(arr(7)),
        NumParse.toInt(arr(8)),
        NumParse.toDouble(arr(9)),
        NumParse.toDouble(arr(10)),
        arr(11),
        arr(12),
        arr(13),
        arr(14),
        arr(15),
        arr(16),
        NumParse.toInt(arr(17)),
        arr(18),
        arr(19),
        NumParse.toInt(arr(20)),
        NumParse.toInt(arr(21)),
        arr(22),
        arr(23),
        arr(24),
        arr(25),
        NumParse.toInt(arr(26)),
        arr(27),
        NumParse.toInt(arr(28)),
        arr(29),
        NumParse.toInt(arr(30)),
        NumParse.toInt(arr(31)),
        NumParse.toInt(arr(32)),
        arr(33),
        NumParse.toInt(arr(34)),
        NumParse.toInt(arr(35)),
        NumParse.toInt(arr(36)),
        arr(37),
        NumParse.toInt(arr(38)),
        NumParse.toInt(arr(39)),
        NumParse.toDouble(arr(40)),
        NumParse.toDouble(arr(41)),
        NumParse.toInt(arr(42)),
        arr(43),
        NumParse.toDouble(arr(44)),
        NumParse.toDouble(arr(45)),
        arr(46),
        arr(47),
        arr(48),
        arr(49),
        arr(50),
        arr(51),
        arr(52),
        arr(53),
        arr(54),
        arr(55),
        arr(56),
        NumParse.toInt(arr(57)),
        NumParse.toDouble(arr(58)),
        NumParse.toInt(arr(59)),
        NumParse.toInt(arr(60)),
        arr(61),
        arr(62),
        arr(63),
        arr(64),
        arr(65),
        arr(66),
        arr(67),
        arr(68),
        arr(69),
        arr(70),
        arr(71),
        arr(72),
        NumParse.toInt(arr(73)),
        NumParse.toDouble(arr(74)),
        NumParse.toDouble(arr(75)),
        NumParse.toDouble(arr(76)),
        NumParse.toDouble(arr(77)),
        NumParse.toDouble(arr(78)),
        arr(79),
        arr(80),
        arr(81),
        arr(82),
        arr(83),
        NumParse.toInt(arr(84))
    ))
    //获得文件是哪种文件，本地或者hdfs
    val configuration = context.hadoopConfiguration
    val fs = FileSystem.get(configuration)
    val path = new Path(ConfigHelper.destPath)
    if (fs.exists(path)){
      fs.delete(path,true)
    }
    //创建DataFrame对象
    val frame: DataFrame = sql.createDataFrame(value,LogSchema.schema)
    //存储数据
    frame.write.parquet(ConfigHelper.destPath)
    //关闭连接
    context.stop()
  }

}
